package org.luxor.commons.elasticsearch.service.impl;

import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.luxor.commons.core.utils.DateUtils;
import org.luxor.commons.core.utils.Jackson2Utils;
import org.luxor.commons.elasticsearch.ElasticClientRest;
import org.luxor.commons.elasticsearch.entity.LogEsPO;
import org.luxor.commons.elasticsearch.entity.LogEsSearchDTO;
import org.luxor.commons.elasticsearch.metadata.EsOrderItem;
import org.luxor.commons.elasticsearch.metadata.EsPageQuery;
import org.luxor.commons.elasticsearch.metadata.EsPageResult;
import org.luxor.commons.elasticsearch.service.ElasticTestService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.io.IOException;
import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;

/**
 * @author Mr.Yan  @date 2019/11/22
 */
@Service
public class ElasticTestServiceImpl implements ElasticTestService {
    private static final Logger logger = LoggerFactory.getLogger(ElasticTestServiceImpl.class);
    /**
     * ElasticTest的索引名称
     */
    String SERVICE_ES_TEST_INDEX = "luxor_framework_logs";

    @Autowired
    private ElasticClientRest elasticClientRest;

    @Override
    public void asyncSave(LogEsPO logEs) throws ElasticsearchException {
        if (logEs == null || logEs.getId() == null) {
            logger.warn("保存 [{}] 索引时,Id或实体为空!", SERVICE_ES_TEST_INDEX);
            return;
        }
        IndexRequest request = new IndexRequest(SERVICE_ES_TEST_INDEX);
        request.id(String.valueOf(logEs.getId()));
        request.source(Jackson2Utils.toJson(logEs), XContentType.JSON);
        elasticClientRest.rest.indexAsync(request, RequestOptions.DEFAULT, new ActionListener<IndexResponse>() {
            @Override
            public void onResponse(IndexResponse updateResponse) {
                logger.debug("write [{}] into elastic succ", updateResponse.getIndex());
            }

            @Override
            public void onFailure(Exception e) {
                logger.debug("write to elastic error log:{}", logEs);
                logger.error("write to elastic error", e);
            }
        });
    }

    @Override
    public void asyncDelete(String id) throws ElasticsearchException {
        DeleteRequest deleteRequest = new DeleteRequest(SERVICE_ES_TEST_INDEX, id);
        elasticClientRest.rest.deleteAsync(deleteRequest, RequestOptions.DEFAULT, new ActionListener<DeleteResponse>() {
            @Override
            public void onResponse(DeleteResponse deleteResponse) {
                logger.debug("delete Doc.id [{}] from elastic.index [{}] succ.", id, SERVICE_ES_TEST_INDEX);
            }

            @Override
            public void onFailure(Exception e) {
                logger.debug("delete Doc.id [{}] from elastic.index [{}] error", id, SERVICE_ES_TEST_INDEX, e);
                logger.error("delete Doc.id [{}] from elastic.index [{}] error", id, SERVICE_ES_TEST_INDEX);
            }
        });
    }

    @Override
    public void asyncUpdate(LogEsPO logEs) throws ElasticsearchException {
        if (logEs == null || logEs.getId() == null) {
            logger.warn("更新 [{}] 索引中的文档时,Id或实体为空!", SERVICE_ES_TEST_INDEX);
            return;
        }
        UpdateRequest request = new UpdateRequest(SERVICE_ES_TEST_INDEX, String.valueOf(logEs.getId()));
        request.doc(Jackson2Utils.toJson(logEs), XContentType.JSON);
        elasticClientRest.rest.updateAsync(request, RequestOptions.DEFAULT, new ActionListener<UpdateResponse>() {
            @Override
            public void onResponse(UpdateResponse updateResponse) {
                logger.debug("update [{}] into elastic succ", updateResponse.getIndex());
            }

            @Override
            public void onFailure(Exception e) {
                logger.debug("update to elastic error log:{}", logEs);
                logger.error("update to elastic error", e);
            }
        });
    }

    @Override
    public LogEsPO getOneByDocId(String docId) throws ElasticsearchException {
        LogEsPO logEs = null;
        try {
            GetRequest getRequest = new GetRequest(SERVICE_ES_TEST_INDEX, docId);
            GetResponse response = elasticClientRest.rest.get(getRequest, RequestOptions.DEFAULT);
            if (response.isExists()) {
                String resJsonText = response.getSourceAsString();
                logEs = Jackson2Utils.toBean(resJsonText, LogEsPO.class);
            }
            return logEs;
        } catch (IOException e) {
            throw new ElasticsearchException("根据文档Id [{}] 获取索引 [{}] 中文档内容时,出现异常:{}", docId, SERVICE_ES_TEST_INDEX, e);
        }
    }

    @Override
    public EsPageResult<LogEsPO> searchPage(EsPageQuery<LogEsSearchDTO> query) throws ElasticsearchException {
        // 分页条件
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.from((query.getCurrent() - 1) * query.getSize());
        searchSourceBuilder.size(query.getSize());

        // 搜索过滤条件
        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
        LogEsSearchDTO condition = query.getSearch();
        if (condition != null) {
            if (condition.getStartTime() != null && condition.getStartTime().getTime() > 0) {
                boolQueryBuilder.must(QueryBuilders.rangeQuery("time").from(condition.getStartTime()));
            }
            if (condition.getEndTime() != null && condition.getEndTime().getTime() > 0) {
                boolQueryBuilder.must(QueryBuilders.rangeQuery("time").to(condition.getEndTime()));
            }
            if (condition.getMessage() != null) {
                boolQueryBuilder.must(QueryBuilders.matchQuery("message", condition.getMessage()));
            }
            searchSourceBuilder.query(boolQueryBuilder);
        }

        // 搜索排序条件
        EsOrderItem orderCondition = query.getOrders();
        if (orderCondition != null) {
            SortOrder order = orderCondition.getAsc() ? SortOrder.ASC : SortOrder.DESC;
            searchSourceBuilder.sort(orderCondition.getColumn(), order);
        }

        // 执行分页查询
        EsPageResult<LogEsPO> pageData = new EsPageResult<>();
        try {
            SearchRequest searchRequest = new SearchRequest(SERVICE_ES_TEST_INDEX);
            // 分页快照保存1分钟
            //searchRequest.scroll(TimeValue.timeValueMinutes(1L));
            searchRequest.source(searchSourceBuilder);
            SearchResponse searchResponse = elasticClientRest.rest.search(searchRequest, RequestOptions.DEFAULT);
            if (RestStatus.OK.equals(searchResponse.status())) {
                // 总记录数
                long total = searchResponse.getHits().getTotalHits().value;
                // 分页数据
                SearchHit[] searchHits = searchResponse.getHits().getHits();
                // 组装分页对象
                pageData.setScrollId(searchResponse.getScrollId());
                pageData.setCurrent(query.getCurrent());
                pageData.setSize(searchHits.length);
                pageData.setTotal(total);
                pageData.setTotalPage(total / query.getSize());
                List<LogEsPO> records = Arrays.stream(searchHits).map(hit ->
                        Jackson2Utils.toBean(hit.getSourceAsString(), LogEsPO.class)
                ).collect(Collectors.toList());
                pageData.setRecords(records);
            }
            return pageData;
        } catch (IOException e) {
            throw new ElasticsearchException("分页查询 [{}] 索引中的文档内容时,出现异常:{}", SERVICE_ES_TEST_INDEX, e);
        }
    }

    @PostConstruct
    public void initIndex() throws ElasticsearchException {
        try {
            // 开始，构建body内容
            XContentBuilder builder = XContentFactory.jsonBuilder();
            builder.startObject();
            {
                builder.startObject("properties");
                {
                    builder.startObject("id");
                    {
                        builder.field("type", "long");
                    }
                    builder.endObject();
                    builder.startObject("userName");
                    {
                        builder.field("type", "keyword");
                        builder.field("ignore_above", 50);
                    }
                    builder.endObject();
                    builder.startObject("ipAddress");
                    {
                        builder.field("type", "ip");
                    }
                    builder.endObject();
                    builder.startObject("traceId");
                    {
                        builder.field("type", "keyword");

                    }
                    builder.endObject();
                    builder.startObject("time");
                    {
                        builder.field("type", "date");
                        builder.field("format", DateUtils.DATE_TIME_PATTERN);
                    }
                    builder.endObject();
                    builder.startObject("level");
                    {
                        builder.field("type", "keyword");
                    }
                    builder.endObject();
                    builder.startObject("application");
                    {
                        builder.field("type", "keyword");
                    }
                    builder.endObject();
                    builder.startObject("message");
                    {
                        builder.field("type", "text");
                    }
                    builder.endObject();
                }
                builder.endObject();
            }
            builder.endObject();
            elasticClientRest.createIndex(SERVICE_ES_TEST_INDEX, builder);
        } catch (IOException e) {
            throw new ElasticsearchException("创建 [{}] 索引时,出现异常:{}", SERVICE_ES_TEST_INDEX, e);
        }
    }

}
